package com.ctfo.ulcp.exp.task

import com.ctfo.ulcp.exp.db.DSL
import com.ctfo.ulcp.exp.db.Tasks
import org.slf4j.LoggerFactory
import java.time.Duration
import java.util.concurrent.*


object LongRunningTaskManager {

    private val taskHolder = ConcurrentHashMap<Int, CompletableFuture<Void>>()
    private val logger = LoggerFactory.getLogger(LongRunningTaskManager::class.java)
    private val executors = ThreadPoolExecutor(4, 16, 0, TimeUnit.MILLISECONDS, ArrayBlockingQueue(30))

    init {
        val monitor = Executors.newScheduledThreadPool(1)
        monitor.scheduleAtFixedRate(this::activePendingTasks, 0, 10, TimeUnit.SECONDS)
        monitor.scheduleAtFixedRate(this::cancelTimeoutTasks, 5, 60, TimeUnit.SECONDS)

    }


    private fun cancelTimeoutTasks() {
        // 在新事务中检查是否存在很久未完成的的并将其状态修改为失败
        DSL.transaction {
            val count = Tasks.updatePendingToFailed(Duration.ofSeconds(60))
            logger.info("修改超时任务状态:{}", count)
        }
    }


    private fun activePendingTasks() {
        logger.info("activePendingTasks, tasks count:{}", taskHolder.size)
        if (taskHolder.isEmpty()) {
            return
        }

        DSL.transaction {
            taskHolder.forEach { (taskId, future) ->
                if (future.isDone) {
                    // 任务已完成，执行清理操作
                    taskHolder.remove(taskId)
                } else {
                    // 记录最近一次活跃时间
                    Tasks.saveTaskHeartbeat(taskId)
                }
            }
        }

    }

    class LongRunningTask(private val taskId: Int, private val runnable: Runnable) : Runnable {
        override fun run() {
            try {
                runnable.run()
                DSL.transaction {
                    Tasks.updateToSucceeded(taskId)
                }

            } catch (e: Throwable) {
                logger.error("execute task failed, taskId:{}, error:", taskId, e)
                DSL.transaction {
                    Tasks.updateToFailed(taskId, e)
                }
            }
        }

    }


    fun submitTask(taskId: Int, runnable: Runnable) {
        val future = CompletableFuture.runAsync(LongRunningTask(taskId, runnable), executors)
        taskHolder[taskId] = future
    }


}

fun main() {
    LongRunningTaskManager.submitTask(10) {
        Thread.sleep(35000)
        throw RuntimeException("test error 1111")
        println("run a task1111111")
    }


    LongRunningTaskManager.submitTask(12) {
        Thread.sleep(26000)
        println("run a task222222")
    }

}